package com.aounhui.productor.listener;

import com.aounhui.common.utils.TypeUtil;
import com.aounhui.productor.common.GetThread;
import com.aounhui.productor.factory.ProductorFactory;
import com.aounhui.productor.service.CommodityService;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.core.ChannelAwareMessageListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.*;

/**
 * @author aounhui
 * @version 1.0
 * @date 2021/6/29 15:42
 * @description 消息监听器，这里有两种思路
 * 第一种批量：就是使用batchingRabbitTemplate
 *      将批量消息压缩为一条消息，然后接收的时候也是接受的压缩消息，再将压缩消息进行解压缩为多条消息，然后处理
 * 第二种是接受所有消息，然后都ack，将信息存到库里或者redis，然后根据情况对信息进行处理，需要reject的消息，这里就需要重新发送消息。
 * 目前这里使用的是第二种解决办法。
 */
@Component
@Slf4j
public class ProductorListener implements ChannelAwareMessageListener {
    @Autowired
    private ProductorFactory productorFactory;
    @Autowired
    private CommodityService commodityService;

    @Override
    public void onMessage(Message message, Channel channel) throws Exception {
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        try {
            //因为传递消息的时候用的map传递,所以将Map从Message内取出需要做些处理
            String msg = message.toString();
//            log.info("收到消息，消息内容为{}", message);
            String[] msgArray = msg.split("'");//可以点进Message里面看源码,单引号直接的数据就是我们的map消息数据
            Map<String, String> msgMap = TypeUtil.turnStringToMap(msgArray[1].trim(),4);
            String type = msgMap.get("type");
            //开启产品生产线
            Thread thread = GetThread.getThreadByName(type);
            if (thread == null || thread.getState().equals(Thread.State.TERMINATED)) {
                new Thread(new CommodityProductThread(type, deliveryTag, channel)).start();
            } else {
                channel.basicReject(deliveryTag, true);
//                log.info("{}生产线正在生产中，此消息目前无法被处理，消息拒绝处理");
            }
        } catch (Exception e) {
            log.error("生产出错，消息被重新放回");
            channel.basicReject(deliveryTag, true);
            e.printStackTrace();
        }
    }

     class CommodityProductThread implements Runnable {
        private String type;
        private long deliveryTag;
        private Channel channel;
        CommodityProductThread(String type, long deliveryTag, Channel channel) {
            this.type = type;
            this.deliveryTag = deliveryTag;
            this.channel = channel;
        }
        @Override
        public void run() {
            try {
                Thread.currentThread().setName(type);
                //产品生产线
                if (commodityService.satisfyCommodity(type)) {
                    log.info("{}已满足最大库存，此次收到生产消息作废", type);
                    channel.basicAck(deliveryTag, true);
                    return;
                }
                productorFactory.productor(type);
                if (!commodityService.satisfyCommodity(type)) {
                    //不满足最大库存量，消息放回，接着处理
                    channel.basicReject(deliveryTag, true);
                    log.info("不满足最大库存，{}生产消息被放回,重新处理", type);
                } else {
                    channel.basicAck(deliveryTag, true);
                    log.info("{}满足最大库存量，消息被正常消费", type);
                }
            }catch (Exception e) {
                log.error("{}生产线故障", type);
                e.printStackTrace();
            }
        }
    }
}
